package com.pai4j.netty.gateway.connector;

import com.alibaba.nacos.api.naming.pojo.Instance;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 网关与Netty节点间的通信连接器
 */
public class NodeConnector {

    /**
     * 节点连接缓存（节点地址 -> Channel）
     */
    private static final ConcurrentHashMap<String, Channel> nodeConnections = new ConcurrentHashMap<>();
    
    private NodeConnector() {}

    /**
     * 发送消息到目标Netty节点
     *
     * @param target 目标节点实例
     * @param message 原始消息内容
     */
    public static void send(Instance target, String message) {
        String nodeKey = target.getIp() + ":" + target.getPort();
        Channel channel = nodeConnections.get(nodeKey);
        
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(message);
        } else {
            // 建立新连接
            createConnection(target, message);
        }
    }

    /**
     * 创建到目标节点的连接并发送消息
     */
    private static void createConnection(Instance target, String message) {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ChannelPipeline pipeline = ch.pipeline();
                     // 解决TCP粘包/半包问题
                     pipeline.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 4, 0, 4));
                     pipeline.addLast(new LengthFieldPrepender(4));
                     
                     // 字符串编解码器
                     pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                     pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                     
                     // 连接状态处理器
                     pipeline.addLast(new ConnectorStateHandler(target));
                 }
             });

            // 同步建立连接
            ChannelFuture future = b.connect(target.getIp(), target.getPort()).sync();
            String nodeKey = target.getIp() + ":" + target.getPort();
            nodeConnections.put(nodeKey, future.channel());
            
            // 发送初始消息
            future.channel().writeAndFlush(message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            group.shutdownGracefully();
        } catch (Exception e) {
            group.shutdownGracefully();
            throw new RuntimeException("Failed to connect to node: " + target, e);
        }
    }

    /**
     * 连接状态处理器（处理节点连接事件）
     */
    private static class ConnectorStateHandler extends ChannelInboundHandlerAdapter {

        private final Instance target;
        
        public ConnectorStateHandler(Instance target) {
            this.target = target;
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            String nodeKey = target.getIp() + ":" + target.getPort();
            nodeConnections.remove(nodeKey);
            ctx.channel().close();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    }
}